/**
 *
 */
package com.raptor.rabbitmq.product;

import com.shihang.core.base.util.ObjectUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;


@Configuration
@EnableConfigurationProperties(RabbitMQProductProperties.class)
@ComponentScan("com.shihang.core.rabbitmq.*")
public class DelegateRabbitMQProducerConfiguration {

    @Autowired
    private RabbitMQProductProperties producerProperties;


    private static final Logger logger = LoggerFactory.getLogger(DelegateRabbitMQProducerConfiguration.class);


    @Bean("connectionFactory")
    public ConnectionFactory connectionFactory() {

        final String host = producerProperties.getHost();
        final int port = producerProperties.getPort();
        final String username = producerProperties.getUsername();
        final String password = producerProperties.getPassword();
        ObjectUtil.checkArgument(host, username, password);
        if (port <= 0) {
            throw new RuntimeException("port can not be zero");
        }
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        return connectionFactory;
    }


    @Bean
    public ProducetSpringProvider producerProvider() {
        ProducetSpringProvider bean = new ProducetSpringProvider();
        return bean;
    }


    @Bean
    @DependsOn("connectionFactory")
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
        // 发送消息时, 设置强制标志, 仅适用于已提供ReturnCallback的情况、
        rabbitTemplate.setMandatory(true);

        /**
         *  如果消息没有到exchange, 则confirm回调, ack=false
         *  如果消息到达exchange, 则confirm回调, ack=true
         *  exchange到queue成功, 则不回调return。
         *  exchange到queue失败, 则回调return(需设置mandatory=true, 否则不回回调, 消息丢失)。
         *
         *  配置错误的 routingKey, 可以触发ReturnCallback()、
         *  rabbitTemplate.convertAndSend("directExchange", "directAAA", msgString, correlationData);
         *  <p>
         *  配置错误的 exchange, 可以触发ConfirmCallback(),并且 ack为false！
         *  rabbitTemplate.convertAndSend("directExchangeBBBBB", "direct", msgString, correlationData);
         */
        // 每个rabbitTemplate只能有一个confirm-callback和return-callback，如果这里配置了，那么写生产者的时候不能再写confirm-callback和return-callback
        // java.lang.IllegalStateException: Only one ConfirmCallback is supported by each RabbitTemplate
//        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
//            if (ack) {
//                logger.info("消息发送成功--> 到达exchange：correlationData({}),cause({})", correlationData, cause);
//            } else {
//                logger.info("消息发送失败--> 未到达exchange：correlationData({}),cause({})", correlationData, cause);
//                // 日志记录, 或者存入Redis、MySQL
//            }
//        });
//        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
//            logger.info("消息丢失：未从exchange发送到queue：exchange({}), routingKey({}), replyCode({}), replyText({}), message:{}", exchange, routingKey, replyCode, replyText, message);
//            // 日志记录, 或者存入Redis、MySQL
//        });
        return rabbitTemplate;
    }


}
